package com.zws.cucumber.bootstrap.cass;

import com.datastax.oss.driver.api.core.AllNodesFailedException;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.CqlSessionBuilder;
import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
import com.zws.cucumber.adaptation.cass.ZwsCqlSessionProxy;
import com.zws.cucumber.bootstrap.ZwsBootstrapEnvsBuilder;
import com.zws.cucumber.util.cass.ZwsCassContext;
import com.zws.cucumber.util.cass.ZwsCassTemplate;
import com.zws.cucumber.util.cass.ZwsCassTemplateImpl;
import com.zws.cucumber.util.cass.backdoor.ZwsCassBackdoorClientImpl;
import org.cassandraunit.CQLDataLoader;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;

import java.io.BufferedReader;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.Collections;

import static com.zws.cucumber.bootstrap.cass.ZwsBootstrapUtil.*;

public abstract sealed class ZwsAbstractCassBootstrap
        implements ZwsCassBootstrap
        permits ZwsStandaloneCassWindowsBootstrap, ZwsStandaloneCassWslBootstrap, ZwsStandaloneCassJenkinsBootstrap {

    private static final String DATACENTER = "datacenter1";

    protected abstract void prepareAndStartCassandra0() throws Exception;

    public final void prepareAndStartCassandra() throws Exception {
        if (ZwsBootstrapUtil.isPortAvailable(getCassandraPort())) {
            prepareAndStartCassandra0();
        }
    }

    public ZwsCassContext boot(String creationScriptPath, String dataScriptPath, String keyspace) throws Exception {
        long timestamp = startTimer();
        try {
            prepareAndStartCassandra();
            return connect(
                    getCassandraPort(),
                    getBackdoorPort(),
                    keyspace,
                    creationScriptPath,
                    dataScriptPath);
        } finally {
            ZwsBootstrapUtil.printf("boot standalone cassandra done, elapsed: %dms\n", System.currentTimeMillis() - timestamp);
        }
    }

    protected ZwsBootstrapEnvsBuilder getPrepareCassandraEnvsBuilder() {
        return ZwsBootstrapEnvsBuilder.builder();
    }

    protected ZwsBootstrapEnvsBuilder getStartCassandraEnvsBuilder() {
        ZwsBootstrapEnvsBuilder builder = ZwsBootstrapEnvsBuilder.builder();
        builder.withEnv("JAVA_HOME", ZwsBootstrapUtil.jdk8Home);
        return builder;
    }

    protected void handleStdoutAndStderr(Process proc) {
        new Thread(() -> {
            try (BufferedReader errorReader = proc.errorReader()) {
                String error;
                while ((error = errorReader.readLine()) != null) {
                    System.err.println(error);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
        new Thread(() -> {
            try (BufferedReader inputReader = proc.inputReader()) {
                String line;
                while ((line = inputReader.readLine()) != null) {
                    System.out.println("standalone cassandra: " + line);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
    }

    protected String waitAndGetPid() {
        try {
            String pathStr = getPathOfSystemLogs();
            Path path = Paths.get(pathStr);
            String content = "";
            for (int i = 1; i <= 120; i++) {
                if (Files.exists(path)) {
                    content = Files.readString(path, StandardCharsets.UTF_8);
                    if (content.contains("Startup complete")) {
                        int idx = content.indexOf("pid: ") + 5;
                        String pid = content.substring(idx, content.indexOf("\n", idx)).trim();
                        println(i + "/120, file found: " + pathStr + ", and cassandra is ready, pid: " + pid);
                        return pid;
                    } else {
                        println(i + "/120, file found: " + pathStr + ", but cassandra not ready yet");
                    }
                } else {
                    println(i + "/120, file not found: " + pathStr);
                }
                Thread.sleep(1_000);
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
        return "";
    }

    protected ZwsCassContext connect(
            int cassandraPort,
            int backdoorPort,
            String keyspace,
            String creationScriptPath,
            String dataScriptPath) throws Exception {

        if (!ZwsBootstrapUtil.isPortAvailable(backdoorPort)) {
            connectBackdoor(backdoorPort);
        }
        CqlSession session = initSession(cassandraPort);
        ZwsCassTemplate template = new ZwsCassTemplateImpl(session);
        ZwsCassContext cassContext = initCassandraServer(cassandraPort, session, template);
        initEmbeddedCassandraServerHelper(session);
        createSchemaAndLoadData(session, template, keyspace, creationScriptPath, dataScriptPath);
        return cassContext;
    }

    protected void connectBackdoor(int backdoorPort) {
        startTimer();
        com.zws.cucumber.bootstrap.ZwsCassBootstrap.cassBackdoorClient = new ZwsCassBackdoorClientImpl(backdoorPort);
        ZwsCqlSessionProxy.asyncExecute(Collections.emptyList(), () -> com.zws.cucumber.bootstrap.ZwsCassBootstrap.cassBackdoorClient.disableTimeout());
        ZwsBootstrapUtil.printf("standalone cassandra: init unsafe truncate client, elapsed: %dms\n", elapsed());
    }

    protected CqlSession initSession(int port) {
        startTimer();
        try {
            CqlSessionBuilder sessionBuilder =
                    CqlSession.builder().addContactPoint(new InetSocketAddress("127.0.0.1", port))
                            .withLocalDatacenter(DATACENTER)
                            .withConfigLoader(DriverConfigLoader.programmaticBuilder()
                                    .withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofSeconds(300))
                                    .build());
            return sessionBuilder.build();
        } catch (AllNodesFailedException e) {
            throw new RuntimeException(e);
        } finally {
            ZwsBootstrapUtil.printf("standalone cassandra: init CqlSession, elapsed: %dms\n", elapsed());
        }
    }

    protected ZwsCassContext initCassandraServer(int port, CqlSession session, ZwsCassTemplate template) {
        startTimer();
        ZwsCassContext cassContext = new ZwsCassContext();
        ZwsBootstrapUtil.setField(ZwsCassContext.class, "isInit", true);
        ZwsBootstrapUtil.setField(ZwsCassContext.class, "cassandraPort", port);
        ZwsBootstrapUtil.setField(ZwsCassContext.class, "session", session);
        ZwsBootstrapUtil.setField(ZwsCassContext.class, "databaseTemplate", template);
        ZwsBootstrapUtil.setField(ZwsCassContext.class, "dataLoader", new CQLDataLoader(session));
        ZwsBootstrapUtil.printf("standalone cassandra: init ZwsCassContext, elapsed: %dms\n", elapsed());
        return cassContext;
    }

    protected void initEmbeddedCassandraServerHelper(CqlSession session) {
        startTimer();
        ZwsBootstrapUtil.setField(EmbeddedCassandraServerHelper.class, "session", session);
        ZwsBootstrapUtil.printf("standalone cassandra: init EmbeddedCassandraServerHelper, elapsed: %dms\n", elapsed());
    }

    protected void createSchemaAndLoadData(CqlSession session, ZwsCassTemplate template, String keyspace, String creationScriptPath, String dataScriptPath) throws Exception {
        startTimer();
        createKeyspaceIfNotExists(session, keyspace);
        template.cleanData(false, keyspace);
        // fixme: to be implemented @ZWS
        // todo: create schema and load data
        ZwsBootstrapUtil.printf("standalone cassandra: clean and load data, elapsed: %dms\n", elapsed());
    }

    public static void createKeyspaceIfNotExists(CqlSession session, String keyspace) {
        long timestamp = System.currentTimeMillis();
        final String cql = "CREATE KEYSPACE IF NOT EXISTS " + keyspace
                + " WITH replication={'class' : 'SimpleStrategy', 'replication_factor':1} AND durable_writes = false";
        session.execute(cql);
        ZwsBootstrapUtil.printf("create keyspace '%s' done, elapsed: %dms\n", keyspace,
                System.currentTimeMillis() - timestamp);
    }
}
